package cn.sheep.dolphin.etl

import ch.hsr.geohash.GeoHash
import cn.sheep.dolphin.common.{DolphinAppComm, Jedispools}
import cn.sheep.dolphin.config.DolphinConfig
import com.alibaba.fastjson.JSON
import org.apache.commons.httpclient.HttpClient
import org.apache.commons.httpclient.methods.GetMethod
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext

import scala.collection.mutable
import scala.util.Random

/** 构建商圈知识库
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/12/3
  */
object BuildBusinessRule {

	def main(args: Array[String]): Unit = {
		if (args.size != 1) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.etl.BuildBusinessRule
				  |<parquetFilePath>
				""".stripMargin)
			sys.exit(-1)
		}

		val Array(parquetFilePath) = args

		val sc = DolphinAppComm.createSparkContext("构建商圈知识库")
		val sQLContext = new SQLContext(sc)

		// 读取parquet文件
		val dataFrame = sQLContext.read.parquet(parquetFilePath)

		// 过滤出合法的经纬度信息
		val filtered = dataFrame
			.select("lat", "long")
			.filter("long<=136 and long>=73 and lat>=3 and lat<=54")
			.distinct()
		dataFrame.map(t => {
			//经度
			val long = t.getAs[String]("long")
			//纬度
			val lat = t.getAs[String]("lat")
			//传递参数经纬度去高德地图官网获取商圈，存入redis

			//获取GeoHash值
			val geoHash = GeoHash.withCharacterPrecision(lat.toDouble, long.toDouble, 8).toBase32
			(geoHash, requestGeocode(lat, long))

			//留下商圈不是空的写入Redis
		}).filter(t=>StringUtils.isNotEmpty(t._2)).foreachPartition(partition => {
			//获取redis连接
			val jedis = Jedispools.getJedis()
			partition.foreach(t => {
				//将结果写进去
				jedis.hset("businessArea", t._1, t._2)
			})
			jedis.close()
		})
		sc.stop()
	}

	/**
		* 调用高德的api接口获取经纬度对应的商圈信息
		*/
	//从高德地图获取信息  https://restapi.amap.com/v3/geocode/regeo?parameters
	def requestGeocode(latLong: (String, String)) = {

		//建立http客户端 相当于浏览器 进而携带参数进行web请求
		val client = new HttpClient()
		var params = Map[String, String](
			"key" -> DolphinConfig._keyList(new Random().nextInt(DolphinConfig._keyList.size)), // 随机key
			"location" -> (latLong._2 + "," + latLong._1)
		).map(kv => kv._1 + "=" + kv._2).mkString("&")

		//获取get方法
		val getMethod = new GetMethod(DolphinConfig._gdReGEOURL + params)

		//执行方法发送get请求 获取返回状态值
		val status = client.executeMethod(getMethod)

		//如果状态值为200 说明有返回值
		if (status == 200) {
			//获取返回内容
			val json = getMethod.getResponseBodyAsString
			//解析json内容 获取商圈信息 LinkedList 或者 null
			val business = parseRequesJson(json)
			//释放连接
			getMethod.releaseConnection()
			//返回商圈信息 LinkedList字符串 或者 null
			business
		} else null
	}

	/**
		* 解析json，获取商圈信息
		* @param json
		* @return
		*/
	def parseRequesJson(json: String) = {
		//解析get请求得到的地理信息json串 成对象
		val positionObj = JSON.parseObject(json)
		//如果状态码为1，进一步解析
		if (positionObj.getIntValue("status") == 1) {
			//得到商圈Array
			val businessAreas = positionObj.getJSONObject("regeocode")
				.getJSONObject("addressComponent")
				.getJSONArray("businessAreas")
			//创建LinkedList  （其特点是增删快，查找慢）
			var linkedList = new mutable.LinkedList[String]()
			for (i <- 0 until businessAreas.size()) {
				//因为商圈信息可能为空，[[]]，但是其为[[]]无法判定为空，所以需要try catch
				try {
					val oneJson = businessAreas.getJSONObject(i)
					//如果商圈为[[]],那么以下会报错
					val name = oneJson.getString("name")
					if (StringUtils.isNotEmpty(name)) {
						linkedList :+= name
					}
				} catch {
					case e: Exception => null
				}
			}
			linkedList.mkString(",")
		} else null
	}
}
